package com.example.bootredis.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.scala.OutputTag;
import org.apache.flink.util.Collector;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @author: xrp
 * @date: 2022/05/05/14:49
 * @description
 */
@Component
public class StartUpComand implements CommandLineRunner {
    @Override
    public void run(String... args) throws Exception {
        

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("118.25.198.145")
                .port(13306)
                .username("root")
                .password("root")
                .databaseList("monomer_order")
                .tableList("monomer_order.article") //多个表逗号分隔
                .deserializer(new LinkConfig()) //自定义返回结果集
                .startupOptions(StartupOptions.initial())
                .serverTimeZone("UTC")
                .build();

        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        // 多表进行分片处理
        OutputTag<String> orderTag = new OutputTag<>("表1", Types.STRING);

        OutputTag<String> userTag = new OutputTag<>("表2", Types.STRING);


        SingleOutputStreamOperator<String> process = streamSource.map((MapFunction<String, JSONObject>) JSON::parseObject).process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context context, Collector<String> collector) {
                if ("article".equals(value.getString("tableName"))) {
                    context.output(orderTag, value.toJSONString());
                } else if ("表2".equals(value.getString("tableName"))) {
                    context.output(userTag, value.toJSONString());
                }
            }
        });
        DataStream<String> orderStream = process.getSideOutput(orderTag);
        DataStream<String> userStream = process.getSideOutput(userTag);
        orderStream.print();
        userStream.print();
        //自定义sink
        streamSource.addSink(new ListenerOrderSink());
        env.executeAsync("fLinkCDC");
    }
}

